package com.atguigu.flink.cep;

import com.atguigu.flink.pojo.OrderEvent;
import com.atguigu.flink.pojo.WaterSensor;
import com.atguigu.flink.utils.MyUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.functions.TimedOutPartialMatchHandler;
import org.apache.flink.cep.nfa.aftermatch.AfterMatchSkipStrategy;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/**
 * Created by Smexy on 2023/2/2
 *
 * 找到超时(30min)和对账失败(只有支付信息没有订单信息)的订单。
 *
 * 34729,create,,1558430842
 * 34729,pay,sd76f87d6,1558430844
 *
 *
 *      四种情况:
 *          1. create --- pay ,未超时   正常
 *          2. create --- pay ,超时。 pay距离create 超过30min  异常
 *          3. create   只下单，不付款。 正常
 *          4. pay    对账失败，没有找到订单下单的信息  异常
 *
    ------------------------------
        定义模式: 业务关系是先create 再支付
            begin(create).optional().within(30min).next(pay)
                    可以匹配 1，  正常匹配
                    可以匹配 2,   超时，只能获取create
                    可以匹配 3,   超时，只能获取create
                目前不能。添加create.optional就可以了。

                一个模式一旦添加了optional,就会变为2个模式。两个模式都会去匹配数据，取匹配的并集。
    ----------------------------------
        假设数据:
 * 34729,create,,1558430842
 * 34729,pay,sd76f87d6,1558430844
 *
 *       有create:   begin(create).within(30min).next(pay)    匹配！  结果 {  34729,create,,1558430842,  34729,pay,sd76f87d6,1558430844   }
 *                    或
         无create:    begin(pay)                             匹配 ！ 结果 { 34729,pay,sd76f87d6,1558430844 }

            通过跳过策略，使一条数据匹配两个模式时，只选最全的那个！

 begin(create,AfterMatchSkipStrategy.skipToNext()).optional().within(30min).next(pay)




 */
public class Demo14_OrderPay
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        WatermarkStrategy<OrderEvent> watermarkStrategy = WatermarkStrategy
            .<OrderEvent>forMonotonousTimestamps()
            .withTimestampAssigner((e, ts) -> e.getEventTime());


        KeyedStream<OrderEvent, Long> ds = env
            .readTextFile("data/OrderLog.csv")
            .map(new MapFunction<String, OrderEvent>()
            {
                @Override
                public OrderEvent map(String value) throws Exception {
                    String[] words = value.split(",");

                    return new OrderEvent(
                        Long.valueOf(words[0]),
                        words[1],
                        words[2],
                        Long.valueOf(words[3]) * 1000
                    );
                }
            })
            .assignTimestampsAndWatermarks(watermarkStrategy)
            .keyBy(OrderEvent::getOrderId);

        /*
            定义模式:
                    begin(create,AfterMatchSkipStrategy.skipToNext()).optional().within(30min).next(pay)
         */
        Pattern<OrderEvent, OrderEvent> pattern = Pattern.<OrderEvent>begin("create", AfterMatchSkipStrategy.skipPastLastEvent())
            .where(new SimpleCondition<OrderEvent>()
            {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "create".equals(value.getEventType());
                }
            })
            .optional()
            .within(Time.minutes(30))
            .next("pay")
            .where(new SimpleCondition<OrderEvent>()
            {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "pay".equals(value.getEventType());
                }
            });


        SingleOutputStreamOperator<String> ds2 = CEP.pattern(ds, pattern)
                                                        .process(new MyProcessTimeOutFunction());

        /*
            定义规则： 从 ②，④，③中匹配 ②和④
                可有: create,pay  匹配 ②
                可无:  pay    匹配④
         */
        Pattern<OrderEvent, OrderEvent> pattern2 = Pattern.<OrderEvent>begin("create", AfterMatchSkipStrategy.skipPastLastEvent())
            .where(new SimpleCondition<OrderEvent>()
            {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "create".equals(value.getEventType());
                }
            })
            .optional()
            .next("pay")
            .where(new SimpleCondition<OrderEvent>()
            {
                @Override
                public boolean filter(OrderEvent value) throws Exception {
                    return "pay".equals(value.getEventType());
                }
            });


        //获取侧流
        KeyedStream<OrderEvent, Long> ds3 = ds2.getSideOutput(new OutputTag<>("timeout", TypeInformation.of(OrderEvent.class)))
                                                   .keyBy(OrderEvent::getOrderId);

        CEP.pattern(ds3,pattern2)
           .select(new PatternSelectFunction<OrderEvent, String>()
           {
               /*
                    可有create匹配:
                            {
                              create = [ OrderEvent(EventType = create) ] ,
                              pay = [ OrderEvent(EventType = pay) ]
                              }

                    没有create匹配:
                             {

                               pay = [ OrderEvent(EventType = pay) ]
                              }

                */
               @Override
               public String select(Map<String, List<OrderEvent>> pattern) throws Exception {
                   Long orderId = pattern.get("pay").get(0).getOrderId();

                   if (pattern.containsKey("create")){
                       return orderId +"超时...";
                   }else{
                       return orderId +"只有pay，缺少create...";
                   }
               }
           })
           .print().setParallelism(1);


        env.execute();




    }

    private static class MyProcessTimeOutFunction extends PatternProcessFunction<OrderEvent, String> implements TimedOutPartialMatchHandler<OrderEvent>
    {

        OutputTag<OrderEvent> outputTag = new OutputTag<>("timeout", TypeInformation.of(OrderEvent.class));

        /*
                处理正常匹配
         */
        @Override
        public void processMatch(Map<String, List<OrderEvent>> match, Context ctx, Collector<String> out) throws Exception {

                    if (!match.containsKey("create")){
                        //输出到侧流
                        ctx.output(outputTag,match.get("pay").get(0));
                    }
        }

        /*
                处理超时匹配
                        { create = [ OrderEvent(EventType = "create") ]   }
         */
        @Override
        public void processTimedOutMatch(Map<String, List<OrderEvent>> match, Context ctx) throws Exception {
                ctx.output(outputTag,match.get("create").get(0));
        }
    }
}
